import { log } from 'node:console';
import fs from 'node:fs';
import { Duplex, Readable, Transform, Writable } from 'node:stream';
import { createGzip } from 'node:zlib';
import { StringDecoder } from 'node:string_decoder';

/*
Node.js 中有四种基本的流类型：
    -Readable​ - 可读的流 (例如 fs.createReadStream()).
    -Writable​ - 可写的流 (例如 fs.createWriteStream()).
    -Duplex​ - 可读写的流 (例如 TCP sockets、zlib streams、crypto streams).
    -Transform​ - 在读写过程中可以修改和变换数据的 Duplex 流 (例如 zlib streams、crypto streams).

可读流事实上工作在下面两种模式之一：flowing 和 paused 。
    -在 flowing 模式下， 可读流自动从系统底层读取数据，并通过 EventEmitter 接口的事件尽快将数据提供给应用。
    -在 paused 模式下，必须显式调用 stream.read() 方法来从流中读取数据片段。
所有初始工作模式为 paused 的 Readable 流，可以通过下面三种途径切换到 flowing 模式：
    -监听 'data' 事件。
    -调用 stream.resume() 方法。
    -调用 stream.pipe() 方法将数据发送到 Writable。
可读流可以通过下面途径切换到 paused 模式：
    -如果不存在管道目标（pipe destination），可以通过调用 stream.pause() 方法实现。
    -如果存在管道目标，可以通过取消 'data' 事件监听，并调用 stream.unpipe() 方法移除所有管道目标来实现。
官方文档（https://www.nodeapp.cn/stream.html#stream_stream）
*/

function testReadStream() {
    let rs = fs.createReadStream('05.jpg');
    let chunks = [];
    rs.on('error', err => log('读取错误'));
    // 设置了编码的话chunk是String类型，否则是Buffer类型
    // 添加了data监听器后将切换成flowing状态并消费数据，所以不要data监听器和pipe()混用，不然可能数据丢失
    rs.on('data', chunk => {
        log('data');
        chunks.push(chunk);
    });
    // 在end中监听ReadStream中的数据读取完成
    rs.on('end', () => {
        let buffer = Buffer.concat(chunks);
        log('end', buffer.length);
    });
}

function testWriteStream() {
    // 实现文件复制
    let rs = fs.createReadStream('05.jpg');
    let ws = fs.createWriteStream('1.jpg');
    rs.pipe(ws);
    // 在finish中监听WriteStream中的数据写入完成
    ws.on('finish', () => {
        log('文件复制完成');
    });
}

function myPipe() {
    // 模拟pipe方法的功能实现文件复制
    let rs = fs.createReadStream('05.jpg');
    let ws = fs.createWriteStream('2.jpg');
    rs.on('data', chunk => {
        let flag = ws.write(chunk);
        if (!flag) {
            rs.pause(); // WriteStream的缓存区已经满了，暂停从ReadStream取数据，flowing -> paused
        }
    });
    rs.on('close', () => {
        ws.end(); // 调用end后才会触发WriteStream的finish回调
    });
    // 如果调用 stream.write(chunk) 方法返回 false，'drain' 事件会在适合恢复写入数据到流的时候触发。
    ws.on('drain', () => {
        log('WriteStream缓存区已清空');
        rs.resume(); // 恢复从ReadStream读取数据，paused -> flowing
    });
    ws.on("finish", () => {
        log('文件复制完成')
    })
}

function testTransformStream(save = false) {
    // 实现对文件内容进行gzip压缩，
    let rs = fs.createReadStream('05.jpg');
    let gs = createGzip()
    // 直接保存压缩文件
    if (save) {
        rs.pipe(gs).pipe(fs.createWriteStream('1.jpg.gz'));
    }
    // 直接获取压缩后的数据
    else {
        rs.pipe(gs)
        let chunks = [];
        gs.on('data', chunk => {
            chunks.push(chunk);
        });
        gs.on('end', () => {
            let buffer = Buffer.concat(chunks);
            log('压缩完成');
        });
    }
}

// 自定义可读流
class MyReadStream extends Readable {
    index = 0;
    _read() {
        if (this.index < 10) {
            this.push(this.index.toString()); // 会自动转换成Buffer类型
            this.index++;
            return;
        }
        this.push(null); // 当push null时流停止
    }
}

function testMyReadStream() {
    let mr = new MyReadStream();
    mr.on('data', data => {
        log(data.toString());
    })
}

// 自定义可写流
class MyWritableStream extends Writable {
    constructor(options) {
        super(options);
        const state = this._writableState;
        this._decoder = new StringDecoder(state.defaultEncoding);
        this.data = '';
    }
    _write(chunk, encoding, callback) {
        if (encoding === 'buffer') {
            chunk = this._decoder.write(chunk);
        }
        this.data += chunk;
        callback();
    }
    // 这个可选的函数将在流关闭之前被调用, 直到callback回调函数执行完成才触发finish事件。
    // 这对于关闭资源或在流结束之前写入缓冲数据很有用。
    _final(callback) {
        this.data += this._decoder.end();
        callback();
    }
}


function testMyWritableStream() {
    const euro = [[0xE2, 0x82], [0xAC]].map(Buffer.from);
    const w = new MyWritableStream();

    w.write('currency: ');
    w.write(euro[0]);
    w.end(euro[1]);

    console.log(w.data); // currency: €
}

// 自定义转换流
class MyTransformStream extends Transform {
    _transform(chunk, encoding, callback) {
        this.push(chunk.toString().toUpperCase());
        callback();
    }
}

function testMyTransformStream() {
    // 获取命令行输入并转换成大写后在命令行输出
    let tf = new MyTransformStream();
    process.stdin.pipe(tf).pipe(process.stdout);
}
